package app.function;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import app.bean.ProductRankResult;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

/**
 * 商品排名处理函数 - 实现实时排名计算
 */
public class ProductRankProcessFunction extends KeyedProcessFunction<String, ProductRankResult, ProductRankResult> {
    
    private final String rankType;
    private final int topN;
    private ListState<ProductRankResult> productListState;
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public ProductRankProcessFunction(String rankType, int topN) {
        this.rankType = rankType;
        this.topN = topN;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        productListState = getRuntimeContext().getListState(
            new ListStateDescriptor<>("productListState", 
                TypeInformation.of(new TypeHint<ProductRankResult>() {}))
        );
    }
    
    @Override
    public void processElement(ProductRankResult value, Context ctx, Collector<ProductRankResult> out) throws Exception {
        // 添加到状态列表
        productListState.add(value);
        
        // 注册定时器，延迟1秒处理排名
        ctx.timerService().registerEventTimeTimer(ctx.timerService().currentWatermark() + 1000);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<ProductRankResult> out) throws Exception {
        // 获取所有商品数据
        List<ProductRankResult> productList = new ArrayList<>();
        for (ProductRankResult product : productListState.get()) {
            productList.add(product);
        }
        
        // 根据排名类型进行排序
        switch (rankType) {
            case "total_sales":
                productList.sort(Comparator.comparing(ProductRankResult::getTotalSalesAmount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "growth_rate":
                productList.sort(Comparator.comparing(ProductRankResult::getGrowthRate, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "visitor_count":
                productList.sort(Comparator.comparing(ProductRankResult::getVisitorCount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "conversion_rate":
                productList.sort(Comparator.comparing(ProductRankResult::getConversionRate, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            default:
                productList.sort(Comparator.comparing(ProductRankResult::getTotalSalesAmount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
        }
        
        // 输出前TopN商品，设置排名和标准字段
        for (int i = 0; i < Math.min(topN, productList.size()); i++) {
            ProductRankResult product = productList.get(i);
            
            // 设置排名信息
            product.setRank(i + 1);
            product.setRankType(rankType);
            
            // 设置实时数仓标准字段
            product.setCurrentTimestamp();
            product.setDataLineage(DATA_SOURCE, VERSION);
            product.setPeriod("1h"); // 1小时统计周期
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long windowStart = timestamp - 3600000; // 1小时前
            product.setWindowTime(sdf.format(new Date(windowStart)), sdf.format(new Date(timestamp)));
            
            out.collect(product);
        }
        
        // 清空状态
        productListState.clear();
    }
}